package com.ustcinfo.study.hbase.xngroup.pengqiang;

import com.ustcinfo.app.DeptMybatisMain;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.client.replication.ReplicationAdmin;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.ibatis.session.SqlSession;
import org.apache.ibatis.session.SqlSessionFactory;
import org.apache.ibatis.session.SqlSessionFactoryBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.io.InputStream;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
public class HBaseOperationDao {
    private static Logger logger = LoggerFactory.getLogger(HBaseOperationDao.class);

    /*集群配置对象*/
    private static Configuration conf = null;

    /*集群连接对象*/
    private static Connection connection = null;

    /*hbase集群管理对象*/
    private static Admin admin = null;

    /*hbase replication管理对象*/
    private static ReplicationAdmin replicationAdmin = null;

    /**
     * 初始化
     */
    static {
        conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.property.clientPort", "2181");
        conf.set("hbase.zookeeper.quorum", "nodex,nodey,nodez");
        try {
            connection = ConnectionFactory.createConnection(conf);
            admin = connection.getAdmin();
            replicationAdmin = new ReplicationAdmin(conf);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 批量插入
     * @param name
     * @param puts
     */
    public static void putList(String name, List<Put> puts) {
        TableName tableName = TableName.valueOf(name);
        try (Table table = connection.getTable(tableName)) {
            for (Put put : puts) {
                table.put(put);
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 单条put
     * @param name
     * @param put
     */
    public static void put(String name, Put put) {
        TableName tableName = TableName.valueOf(name);
        try (Table table = connection.getTable(tableName)) {
            table.put(put);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public static void main(String[] args) {
        List<TransferTaskTo> transferTaskTos = MybatisTest.selectAll();
        for (TransferTaskTo transferTaskTo : transferTaskTos){
            Put put = new Put(transferTaskTo.getId().toString().getBytes());
            if(transferTaskTo.getId()!= null)
            put.addColumn("F1".getBytes(), "id".getBytes(), transferTaskTo.getId().toString().getBytes());
            if(transferTaskTo.getTaskId()!= null)
            put.addColumn("F1".getBytes(), "task_id".getBytes(), transferTaskTo.getTaskId().getBytes());
            if(transferTaskTo.getCreateTime()!= null)
            put.addColumn("F1".getBytes(), "create_time".getBytes(), transferTaskTo.getCreateTime().toString().getBytes());
            if(transferTaskTo.getStartTime()!= null)
            put.addColumn("F1".getBytes(), "start_time".getBytes(), transferTaskTo.getStartTime().toString().getBytes());
            if(transferTaskTo.getOutFilterNumber()!= null)
            put.addColumn("F1".getBytes(), "out_filter_number".getBytes(), transferTaskTo.getOutFilterNumber().getBytes());
            if(transferTaskTo.getFromRunParam()!= null)
            put.addColumn("F1".getBytes(), "from_run_param".getBytes(), transferTaskTo.getFromRunParam().getBytes());
            if(transferTaskTo.getCurrentRetryTimes()!= null)
            put.addColumn("F2".getBytes(), "current_retry_times".getBytes(), transferTaskTo.getCurrentRetryTimes().getBytes());
            if(transferTaskTo.getRetryFlag()!= null)
            put.addColumn("F2".getBytes(), "retry_flag".getBytes(), transferTaskTo.getRetryFlag().getBytes());
            if(transferTaskTo.getParentTaskId()!= null)
            put.addColumn("F2".getBytes(), "parent_task_id".getBytes(), transferTaskTo.getParentTaskId().getBytes());
            if(transferTaskTo.getEndTime()!= null)
            put.addColumn("F2".getBytes(), "end_time".getBytes(), transferTaskTo.getEndTime().toString().getBytes());
            if(transferTaskTo.getKillTime()!= null)
            put.addColumn("F2".getBytes(), "kill_time".getBytes(), transferTaskTo.getKillTime().toString().getBytes());
            if(transferTaskTo.getFailureTime()!= null)
            put.addColumn("F2".getBytes(), "failure_time".getBytes(), transferTaskTo.getFailureTime().toString().getBytes());
            put("BD_TEST", put);
        }
    }
}

class MybatisTest {
    /**
     * 获取表所有数据，展示list列表
     */
    public static List<TransferTaskTo> selectAll(){
        SqlSession sqlSession = getSsion();
        String statement = "com.ustcinfo.mapping.TransferTaskToMapper.selectAll";
        List<TransferTaskTo> transferTaskTos = sqlSession.selectList(statement);
        return transferTaskTos;
    }
    /**
     * 根据ID查询表，展示一条数据
     * @param key  查询ID
     */
    public static TransferTaskTo selectByKey(Integer key){
        SqlSession sqlSession = getSsion();
        String statement = "com.ustcinfo.mapping.TransferTaskToMapper.selectByPrimaryKey";
        TransferTaskTo transferTaskTo = sqlSession.selectOne(statement,key);
        return transferTaskTo;
    }

    private static SqlSession getSsion(){
        //mybatis的配置文件
        String resource = "mybatis.cfg.xml";
        //使用类加载器加载mybatis的配置文件（它也加载关联的映射文件）
        InputStream is = DeptMybatisMain.class.getClassLoader().getResourceAsStream(resource);
        //构建sqlSession的工厂
        SqlSessionFactory sessionFactory = new SqlSessionFactoryBuilder().build(is);
        //使用MyBatis提供的Resources类加载mybatis的配置文件（它也加载关联的映射文件）
        //Reader reader = Resources.getResourceAsReader(resource);
        //构建sqlSession的工厂
        //SqlSessionFactory sessionFactory = new SqlSessionFactoryBuilder().build(reader);
        //创建能执行映射文件中sql的sqlSession
        SqlSession session = sessionFactory.openSession();
        return session;
    }
}